/*
 * Copyright (c) 2014, Hsu Shih-Chia (schsu01@gmail.com)
 * All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.qood;

import java.io.*;
import java.util.*;

/**
 * QDB is a collection of named QModel, like a collection of named Tables in a RDBMS.
 * <p/>
 * <p>You can execute relational query by {@link #apply(QSQL)}
 *
 * @author SCHsu01
 * @version 1.0
 */
public interface QDB extends Serializable, Comparable {
  /**
   * Add or Set a named QModel
   *
   * @param name  alias
   * @param model actual data
   * @return a reference to this Object
   */
  QDB with(String name, QModel model);

  /**
   * Add or Set a collection of named QModel.
   * <p>with("X", QDB{A,B}) = with("X.A,~).with("X.B",~)
   *
   * @param name alias
   * @param qdb  a collection of named QModel
   * @return a reference to this Object
   */
  QDB with(String name, QDB qdb);

  /**
   * Get names of all QModel in this QDB
   *
   * @return all names
   */
  String[] getNames();

  /**
   * Get QModel data by specified name
   *
   * @param name alias
   * @return found QModel
   */
  QModel getModel(String name);

  /**
   * Prints data to System.out for debugging, prints all if rowCnt &lt; 0.
   *
   * @param rowCnt the rows to be printed out
   * @return a reference to this Object
   */
  QDB debug(int rowCnt);

  /**
   * apply a QSQL to a QView
   *
   * @param query predefined QSQL
   * @return QView which is about to be executed into another QModel as result
   */
  QView apply(QSQL query);
}

final class DB implements QDB {
  final Map<String, Model> models;

  public DB(Map<String, Model> models) {
    this.models = null == models ? new LinkedHashMap<String, Model>() : new LinkedHashMap<String, Model>(models);
  }

  public String toString() {
    return getClass().getSimpleName() + models;
  }

  public String[] getNames() {
    return StringUtil.toArray(models.keySet());
  }

  public QModel getModel(String name) {
    return models.get(name);
  }

  public QDB with(String name, QModel model) {
    if (null == model) models.remove(name);
    else models.put(name, (Model) model);
    return this;
  }

  public QDB with(String name, QDB qdb) {
    for (final Map.Entry<String, Model> entry : ((DB) qdb).models.entrySet())
      with(name + "." + entry.getKey(), entry.getValue());
    return this;
  }

  public QView apply(QSQL query) {
    return new View(models, (SQL) (query instanceof View ? ((View) query).getSQL() : query));
  }

  public QDB debug(int rowCnt) {
    System.out.println(getClass().getSimpleName() + " contains " + models.size() + " models:");
    for (final Map.Entry<String, Model> entry : models.entrySet()) {
      System.out.println(entry);
      entry.getValue().debug(rowCnt);
    }
    return this;
  }

  protected final Object writeReplace() throws ObjectStreamException {
    return new Q(this);
  }

  public int compareTo(Object o) {
    if (o instanceof QDB) {
      final String[] his = ((QDB) o).getNames();
      int n = models.size(), c;
      if (0 != (c = n - his.length)) return c;
      final String[] mine = getNames();
      for (int i = 0; i < n; i++) if (0 != (c = mine[i].compareTo(his[i]))) return c;
      return 0;
    } else return toString().compareTo(String.valueOf(o));
  }
}

final class ModelQFS extends Model {
  private final String pathQFS;
  private final int buf, rows, cols, bufMax;
  private final long[] pos;
  private final Class[] type;
  private final transient C[] c;

  ModelQFS(String pathQFS, int buf, int rows, String[] name, Class[] type, IO[] io, long[] pos) {
    super(name);
    this.pathQFS = pathQFS;
    this.buf = buf;
    this.rows = rows;
    this.type = type;
    this.bufMax = (rows - 1) / buf;//bufCnt = bufMax + 1
    this.pos = pos;
    final int n = this.cols = getColCount();
    final Col[] cols = this.c = new C[n];
    for (int i = 0; i < n; i++) cols[i] = new C(i, io[i]);
  }

  private static synchronized ObjectInput loadHeader(String pathQFS) throws IOException {
    BufferInput ri = null;
    try {//mark synchronized to prevent multiple file reading
      ri = new BufferInput(pathQFS);
      final int ver = ri.readInt();//1.ver
      if (BuilderQFS.VER != ver) throw new NotSerializableException("Unknown version: " + ver);
      final long pos = ri.readLong();//2.skipPos (... 3.data ...)
      return ri.readObjects(pos, ri.fileLength());
    } finally {
      if (null != ri) try {
        ri.close();
      } catch (IOException ignored) {/*do nothing*/}
    }
  }

  private static synchronized ObjectInput loadRows(String pathQFS, long pos, long next) throws IOException {
    BufferInput ri = null;
    try {//mark synchronized to prevent multiple file reading
      return (ri = new BufferInput(pathQFS)).readObjects(pos, next);
    } finally {
      if (null != ri) try {
        ri.close();
      } catch (IOException ignored) {/*do nothing*/}
    }
  }

  static QModel getInstance(String pathQFS) {
    try {
      final ObjectInput oi = loadHeader(pathQFS);
      final int buf = oi.readInt(), rows = oi.readInt(), cols = oi.readInt();//4.buffer 5.rows 6.cols
      final String[] colName = new String[cols];
      final IO[] io = new IO[cols];
      final Class[] type = new Class[cols];
      for (int i = 0; i < cols; i++) {
        colName[i] = (String) oi.readObject();//7.colName
        type[i] = (io[i] = IO.getInstance(oi.readByte())).readType(oi);//8.colType
      }
      final int posCnt = rows > 0 ? cols * (1 + (rows - 1) / buf) + 1 : 0;
      final long[] posAll = new long[posCnt];//9.posAll
      for (int i = 0; i < posCnt; i++) posAll[i] = oi.readLong();
      return new ModelQFS(pathQFS, buf, rows, colName, type, io, posAll);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  protected Object writeReplace() throws ObjectStreamException {
    if (rows > buf) throw new NotSerializableException("too many to serialize " + rows + " rows");
    return super.writeReplace();
  }

  Col[] getCols() {
    return c;
  }

  public int getRowCount() {
    return rows;
  }

  public Object getValue(int row, int col) {
    return c[col].getRow(row);
  }

  private final class C extends Col {
    private final int colId;
    private final IO io;
    private final Object[] value;
    private int start;

    C(int colId, IO io) {
      this.colId = colId;
      this.io = io;
      this.start = -buf;
      this.value = io.create(buf, getType());
    }

    @SuppressWarnings("unchecked")
    Object getRow(int row) {
      int index = row - start;
      if (index < 0 || index >= buf) synchronized (this) {//out of range, reload
        final int idx = row / buf;
        index = row - (start = idx * buf);//update start and index
        final int id = idx * cols + colId, rowCnt = idx == bufMax ? rows - start : buf;
//        System.out.println(name[colId] + " read " + start + "+" + rowCnt);
        try {
          io.readValues(loadRows(pathQFS, pos[id], pos[id + 1]), rowCnt, value);
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      }
      return value[index];
    }

    Class getType() {
      return type[colId];
    }

    int expectSize() {
      return rows;
    }

    @Override
    @SuppressWarnings({"unchecked"})
    Col toPartial(IntRead rows, Map<IntRead, IntRead> map) {
      if (rows.isSimple()) return this;
      final int n = rows.size();
      if (n > buf) return super.toPartial(rows, map);//wrap
      final Class type = getType();
      final Object[] values = io.create(n, type);
      for (int i = 0; i < n; i++) {
        final int row = rows.get(i);
        values[i] = row >= 0 ? getRow(row) : null;//slow if random access many files
      }
      return getValues(type, values);
    }
  }
}

/**
 * surrogate for serialization
 */
final class Q implements Serializable {
  private static final long serialVersionUID = 0L;
  private static final int MODEL = 1, DB = 2;//zip is much slower!
  private transient int version;
  private transient Model model;
  private transient DB db;

  Q(Model model) {
    this.model = model;
    this.version = MODEL;
  }

  Q(DB db) {
    this.db = db;
    this.version = DB;
  }

  private Object readResolve() throws ObjectStreamException {
    switch (version) {
      case MODEL:
        return model;
      case DB:
        return db;
      default:
        throw unknownVersion();
    }
  }

  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
    final int version = this.version = in.readInt();//read version
    switch (version) {
      case MODEL:
        readModel(in);
        break;
      case DB:
        readDB(in);
        break;
      default:
        throw unknownVersion();
    }
  }

  private void writeObject(ObjectOutputStream out) throws IOException {
    out.writeInt(version);//write version
    switch (version) {
      case MODEL:
        writeModel(out);
        break;
      case DB:
        writeDB(out);
        break;
      default:
        throw unknownVersion();
    }
  }

  private NotSerializableException unknownVersion() {
    return new NotSerializableException("Unknown version: " + version);
  }

  private void readDB(ObjectInput in) throws IOException, ClassNotFoundException {
    final Map<String, Model> models = new LinkedHashMap<String, Model>();
    final int count = in.readInt();
    for (int i = 0; i < count; i++) {
      readModel(in);//set this.model
      models.put((String) in.readObject(), model);
    }
    db = new DB(models);
  }

  @SuppressWarnings({"unchecked"})
  private void readModel(ObjectInput in) throws IOException, ClassNotFoundException {
    final int rows = in.readInt(), cols = in.readInt();//1.row 2.col
    final String[] name = new String[cols];
    for (int i = 0; i < cols; i++) name[i] = (String) in.readObject();//3.names
    final Col[] col = new Col[cols];//for each cols, 4.typeId 5.[ClsName] 6.values
    for (int i = 0; i < cols; i++) col[i] = IO.getInstance(in.readByte()).readCol(in, rows);
    model = new ModelCol(rows, col, name);
  }

  private void writeModel(ObjectOutput out) throws IOException {
    final int rows = model.getRowCount();
    out.writeInt(rows);//1.row
    out.writeInt(model.getColCount());//2.col
    for (final String name : model.getColNames()) out.writeObject(name);//3.names
    for (final Col col : model.getCols()) IO.getInstance(col).writeCol(out, rows, col);//4.typeId 5.[ClsName] 6.values
  }

  private void writeDB(ObjectOutput out) throws IOException {
    final Map<String, Model> models = db.models;
    out.writeInt(models.size());
    for (final Map.Entry<String, Model> entry : models.entrySet()) {
      model = entry.getValue();
      writeModel(out);
      out.writeObject(entry.getKey());
    }
  }
}

abstract class IO<T> {//try to output primitive types than Objects
  private static final Map<Class, IO> MAP = new HashMap<Class, IO>();
  private static final List<IO> LIST = new ArrayList<IO>();
  private static final IO CONSTANT = new Default() {//static IO's order is important!
    @Override
    @SuppressWarnings({"unchecked"})
    Col readCol(ObjectInput in, int rows) throws IOException, ClassNotFoundException {
      return Col.getConst(readType(in), read(in));//read once
    }

    @Override
    void writeCol(ObjectOutput out, int rows, Col col) throws IOException {
      writeType(out, col);
      write(out, col.getRow(0));//write once
    }
  };
  private static final IO DEFAULT = new Default();
  private static final IO BOOLEAN = new IO<Boolean>(Boolean.class) {
    Boolean[] create(int rows, Class type) {
      return new Boolean[rows];
    }

    Boolean read(ObjectInput in) throws IOException {
      return in.readBoolean();
    }

    void write(ObjectOutput out, Boolean value) throws IOException {
      out.writeBoolean(value);
    }
  };
  private static final IO BYTE = new IO<Byte>(Byte.class) {
    Byte[] create(int rows, Class type) {
      return new Byte[rows];
    }

    Byte read(ObjectInput in) throws IOException {
      return in.readByte();
    }

    void write(ObjectOutput out, Byte value) throws IOException {
      out.writeByte(value);
    }
  };
  private static final IO SHORT = new IO<Short>(Short.class) {
    Short[] create(int rows, Class type) {
      return new Short[rows];
    }

    Short read(ObjectInput in) throws IOException {
      return in.readShort();
    }

    void write(ObjectOutput out, Short value) throws IOException {
      out.writeShort(value);
    }
  };
  private static final IO CHAR = new IO<Character>(Character.class) {
    Character[] create(int rows, Class type) {
      return new Character[rows];
    }

    Character read(ObjectInput in) throws IOException {
      return in.readChar();
    }

    void write(ObjectOutput out, Character value) throws IOException {
      out.writeChar(value);
    }
  };
  private static final IO INT = new IO<Integer>(Integer.class) {
    Integer[] create(int rows, Class type) {
      return new Integer[rows];
    }

    Integer read(ObjectInput in) throws IOException {
      return in.readInt();
    }

    void write(ObjectOutput out, Integer value) throws IOException {
      out.writeInt(value);
    }
  };
  private static final IO FLOAT = new IO<Float>(Float.class) {
    Float[] create(int rows, Class type) {
      return new Float[rows];
    }

    Float read(ObjectInput in) throws IOException {
      return in.readFloat();
    }

    void write(ObjectOutput out, Float value) throws IOException {
      out.writeFloat(value);
    }
  };
  private static final IO DOUBLE = new IO<Double>(Double.class) {
    Double[] create(int rows, Class type) {
      return new Double[rows];
    }

    Double read(ObjectInput in) throws IOException {
      return in.readDouble();
    }

    void write(ObjectOutput out, Double value) throws IOException {
      out.writeDouble(value);
    }
  };
  private static final IO LONG = new IO<Long>(Long.class) {
    Long[] create(int rows, Class type) {
      return new Long[rows];
    }

    Long read(ObjectInput in) throws IOException {
      return in.readLong();
    }

    void write(ObjectOutput out, Long value) throws IOException {
      out.writeLong(value);
    }
  };
  private static final IO STRING = new IO<String>(String.class) {
    String[] create(int rows, Class type) {
      return new String[rows];
    }

    String read(ObjectInput in) throws IOException, ClassNotFoundException {
      return (String) in.readObject();
    }

    void write(ObjectOutput out, String value) throws IOException {
      out.writeObject(value);
    }
  };
  private static final IO DATE = new IO<Date>(Date.class, true) {
    Date[] create(int rows, Class type) {
      return new Date[rows];
    }

    Date read(ObjectInput in) throws IOException {
      final long time = in.readLong();
      return Long.MIN_VALUE == time ? null : new Date(time);
    }

    void write(ObjectOutput out, Date value) throws IOException {
      out.writeLong(null == value ? Long.MIN_VALUE : value.getTime());
    }
  };
  private final Class<T> type;
  private final int index;
  private final boolean nullHandled;

  private IO(Class<T> type, boolean nullHandled) {
    this.type = type;
    this.nullHandled = nullHandled;
    this.index = LIST.size();
    LIST.add(this);
    if (null != type) MAP.put(type, this);
  }

  private IO(Class<T> type) {
    this(type, false);
  }

  static IO getInstance(Class type) {
    final IO got = MAP.get(type);
    return null == got ? DEFAULT : got;
  }

  static IO getInstance(Col data) {
    return data.isConstant() ? CONSTANT : getInstance(data.getType());
  }

  static IO getInstance(byte index) {
    return LIST.get(index);
  }

  Class<T> readType(ObjectInput in) throws IOException, ClassNotFoundException {
    return type;
  }

  void writeType(ObjectOutput out, Col col) throws IOException {//for Serialize
    writeType(out, (Class) null);
  }

  void writeType(ObjectOutput out, Class type) throws IOException {//for Builder
    out.writeByte(index);//without type object
  }

  abstract T[] create(int rows, Class type);

  abstract T read(ObjectInput in) throws IOException, ClassNotFoundException;

  abstract void write(ObjectOutput out, T value) throws IOException;

  @SuppressWarnings({"unchecked"})
  Col readCol(ObjectInput in, int rows) throws IOException, ClassNotFoundException {
    final Class<T> type = readType(in);
    final T[] values = create(rows, type);
    readValues(in, rows, values);
    return Col.getValues(type, values);
  }

  void readValues(ObjectInput in, int rows, T[] values) throws IOException, ClassNotFoundException {
    final int nulls;//null cnt
    if (nullHandled || (0 == (nulls = in.readInt()))) {
      for (int i = 0; i < rows; i++) values[i] = read(in);
    } else if (nulls <= (rows >> 1)) {//nulls <= half rows
      final int[] nullId = new int[nulls];
      for (int i = 0; i < nulls; i++) nullId[i] = in.readInt();
      for (int row = 0, i = 0, id = nullId[i]; row < rows; row++)
        if (id == row) id = ++i == nulls ? rows : nullId[i];//skip null index
        else values[row] = read(in);//read non-null index
    } else {
      final int nonNulls = rows - nulls;
      final int[] nonNullId = new int[nonNulls];
      for (int i = 0; i < nonNulls; i++) nonNullId[i] = in.readInt();
      for (int i = 0; i < nonNulls; i++) values[nonNullId[i]] = read(in);//read non-null index
    }
  }

  @SuppressWarnings("unchecked")
  void writeCol(ObjectOutput out, int rows, Col col) throws IOException {
    writeType(out, col);
    if (nullHandled) for (int i = 0; i < rows; i++) write(out, (T) col.getRow(i));
    else {
      final IntList nullId = IntList.create(/*unknown size*/);
      final T[] temp = create(rows, col.getType());
      int nonNulls = 0;
      T value;
      for (int i = 0; i < rows; i++)
        if (null == (value = (T) col.getRow(i))) nullId.add(i);
        else temp[nonNulls++] = value;
      writeNulls(out, rows, nullId);
      for (int i = 0; i < nonNulls; i++) write(out, temp[i]);
    }
  }

  @SuppressWarnings("unchecked")
  void writeValues(ObjectOutput out, int rows, Object[] values) throws IOException {
    if (nullHandled) for (int i = 0; i < rows; i++) write(out, (T) values[i]);
    else {
      final IntList nullId = IntList.create(/*unknown size*/);
      int nonNulls = 0;
      Object value;
      for (int i = 0; i < rows; i++)
        if (null == (value = values[i])) nullId.add(i);
        else values[nonNulls++] = value;//may be the same value
      writeNulls(out, rows, nullId);
      for (int i = 0; i < nonNulls; i++) write(out, (T) values[i]);
    }
  }

  private void writeNulls(ObjectOutput out, int rows, IntList nullId) throws IOException {
    final int nulls = nullId.size();
    out.writeInt(nulls);//null cnt
    if (nulls <= (rows >> 1)) //nulls <= half rows
      for (int i = 0; i < nulls; i++) out.writeInt(nullId.get(i));//write null index
    else for (int row = 0, i = 0, id = nullId.get(i); row < rows; row++)
      if (id == row) id = ++i == nulls ? rows : nullId.get(i);//skip null index
      else out.writeInt(row);//write non-null index
  }

  private static class Default extends IO {
    @SuppressWarnings("unchecked")
    private Default() {
      super(null, true);
    }

    @Override
    final Class readType(ObjectInput in) throws IOException, ClassNotFoundException {
      final String name = (String) in.readObject();
      return Class.forName(name);
    }

    @Override
    final void writeType(ObjectOutput out, Col col) throws IOException {
      writeType(out, col.getType());
    }

    @Override
    final void writeType(ObjectOutput out, Class type) throws IOException {
      super.writeType(out, type);//writeByte(0)
      out.writeObject(type.getName());
    }

    final Object[] create(int rows, Class type) {
      return Comparable.class.isAssignableFrom(type) ? new Comparable[rows] : new Object[rows];
    }

    final Object read(ObjectInput in) throws IOException, ClassNotFoundException {
      return in.readObject();
    }

    final void write(ObjectOutput out, Object value) throws IOException {
      out.writeObject(value);
    }
  }
}

final class BufferOutput extends ByteArrayOutputStream {
  private ObjectOutputStream oos;

  BufferOutput(int size) {
    super(size);
  }

  ObjectOutputStream newOOS() throws IOException {
    return oos = new ObjectOutputStream(this);
  }

  BufferOutput writeInt(int... values) throws IOException {
    for (final int value : values) oos.writeInt(value);
    return this;
  }

  BufferOutput writeLong(long... values) throws IOException {
    for (final long value : values) oos.writeLong(value);
    return this;
  }

  long writeTo(RandomAccessFile raf) throws IOException {
    final long pos = raf.getFilePointer();
    oos.close();
    oos = null;
    raf.write(buf, 0, count);
    reset();
    return pos;
  }
}

final class BufferInput {
  private final RandomAccessFile file;

  BufferInput(String filePath) throws IOException {
    this.file = new RandomAccessFile(filePath, "r");
  }

  public void close() throws IOException {
    file.close();
  }

  public int readInt() throws IOException {
    return file.readInt();
  }

  public long readLong() throws IOException {
    return file.readLong();
  }

  public long fileLength() throws IOException {
    return file.length();
  }

  public ObjectInput readObjects(long pos, long next) throws IOException {
    file.seek(pos);
    final int size = (int) (next - pos);
    final byte[] buffer = new byte[size];
    return file.read(buffer) > 0 ? new ObjectInputStream(new ByteArrayInputStream(buffer)) : null;
  }
}